package day7

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, IntegerType, StructType}

class CustonSum extends UserDefinedAggregateFunction{
  // 1、自定义的聚合函数，有几个参数，以及参数的类型
  override def inputSchema: StructType = new StructType().add("age",IntegerType)

  // 2、缓冲区中要定义的变量有几个，因为我们要完成的需求是求和，所以需要定义一个变量作为累加器
  override def bufferSchema: StructType = new StructType().add("sum",IntegerType)

  // 3、自定义的聚合函数，函数的返回值类型
  override def dataType: DataType = IntegerType

  // 4、相同的输入是否有相同的输出，如果写true,则代表有，否则没有
  override def deterministic: Boolean = true

  // 初始化
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0
  }

  // 参数1：累加器  参数2：sql调用时，输入的数据
  // 将缓冲区的数据加上输入的数据，然后更新到缓冲区中
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getInt(0) + input.getInt(0)   // 相当于 sum = sum + i
    println("buffer = " + buffer)
  }

  // 累加器（用于分区间的累加）合并缓冲区的数据 参数二：各分区的结果
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getInt(0) + buffer2.getInt(0)
//    println("buffer1 = " + buffer1)
  }

  // 该方法的返回值会成为你自定义函数的返回值
  override def evaluate(buffer: Row): Any = {
    println("buffer = " + buffer)
    buffer.getInt(0)
  }

}
